package cn.doitedu.test;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MeterView;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Flink的度量API
 * 统计处理数据的条数、处理数据的时间、每秒处理的数据的条数
 */

/**
 * 可视化报表工具：grafana，只要配置地址即可
 * <p>
 * 获取flink统计度量的 resi api 地址
 * # GET /v1/jobs/<jobid>/vertices/<vertexid>/subtasks/metrics
 * http://localhost:8081/v1/jobs/1bd848cfcfbc02243b7091dc71716f71/vertices/cbc357ccb763df2852fee8c4fc7d55f2/subtasks/metrics/
 * http://localhost:8081/v1/jobs/1bd848cfcfbc02243b7091dc71716f71/vertices/cbc357ccb763df2852fee8c4fc7d55f2/subtasks/metrics?get=Process.g1.eleCount
 */


public class MetricDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //env.setParallelism(1);
        //1
        //2
        //a
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<String> res = lines.process(new ProcessFunction<String, String>() {

            private Counter allCounter;
            private Counter oddCounter;
            private Counter evenCounter;
            private Counter strCounter;

            private MeterView meterView;

            @Override
            public void open(Configuration parameters) throws Exception {
                //定义度量（计数器）
                MetricGroup metricGroup = getRuntimeContext().getMetricGroup().addGroup("doit");
                oddCounter = metricGroup.counter("odd");
                evenCounter = metricGroup.counter("even");
                strCounter = metricGroup.counter("str");
                allCounter = metricGroup.counter("all");

                meterView = metricGroup.meter("rate", new MeterView(allCounter, 10));

            }

            @Override
            public void processElement(String in, Context ctx, Collector<String> out) throws Exception {
                allCounter.inc();
                try {
                    int num = Integer.parseInt(in);
                    if (num % 2 == 0) {
                        evenCounter.inc();
                    } else {
                        oddCounter.inc();
                    }
                } catch (NumberFormatException e) {
                    //e.printStackTrace();
                    strCounter.inc();
                }
                out.collect(in);
            }
        });

        res.print();

        env.execute();


    }
}
